package com.bungarus.busi;

import com.bungarus.cluster.NodeInfo;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.AbstractBungarusServer;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageType;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.apache.log4j.Logger;
import org.springframework.data.mongodb.core.MongoTemplate;

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by tang on 2018/11/29.
 */
public class TransferInnerClient extends InnerClient{
    Logger logger = Logger.getLogger(TransferInnerClient.class);

    public TransferInnerClient(AbstractBungarusServer hostedServer, int index, CopyOnWriteArrayList<NodeInfo> queue, MongoTemplate mongoTemplate) {
        super(hostedServer, index, queue, mongoTemplate);
    }

    @Override
    public void sendMessage(MessageProtobuf.Message msg, String clusterTag, ChannelHandlerContext forAck) throws BungarusClientException {
        Channel connectionChannel = this.clientChannels.get(clusterTag);
        if(null == connectionChannel || !connectionChannel.isActive()) {
            throw new BungarusClientException("no channel can be used to send message.");
        }

        MessageProtobuf.Message newMsg ;
        if(msg.getHeader().getType() == MessageType.CLONE) {
            newMsg = msg;
        }
        else {
            newMsg = MessageBuilder.cloneMessageWithNewId(msg);
        }

        channel.writeAndFlush(newMsg).addListener(channelFuture -> {
            //after sending message, the message is putted into a queue to wait for being acknowledged
            if(channelFuture.isDone() && channelFuture.isSuccess()) {
                if(null != forAck) {
                    forAck.writeAndFlush(MessageBuilder.buildAckMessage(msg.getHeader().getId(), msg.getHeader().getId2()));
                }
                DischargeQueue dischargeQueue = Discharger.getQueueInstance();
                dischargeQueue.addToQueue(newMsg);
            }
            else {//retry
                Long id = Long.valueOf(newMsg.getHeader().getId());
                int e;
                if(null == retryQueue.get(id)){
                    AtomicInteger ai = new AtomicInteger(RETRY_INIT_COUNT);
                    e = ai.getAndIncrement();
                    retryQueue.put(id, ai);
                }
                else {
                    e = retryQueue.get(id).getAndIncrement();
                    if(e > RETRY_TOTAL_COUNT) {
                        //save the message into persistent storage (supported by mongo)
                        saveTryFailedMessage(newMsg);
                        ///////////////////////////////////////////////////////////
                        try {
                            channel.close();
                            this.clientChannels.remove(clusterTag);
                            this.connectNextNode(this.eventLoopGroup);
                        }
                        catch (BungarusClientException e1){
                            this.reLoadRemoteServers(this.eventLoopGroup);
                        }
                        finally {
                            retryQueue.remove(id);
                        }
                        return;
                    }
                }
                Thread.sleep((long)Math.pow(2, e) * 1000);
                sendMessage(newMsg, clusterTag, forAck);
            }
        });
    }
}
